package com.example;

import com.example.config.KafkaConfig;
import com.example.model.CanalMessage;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.File;
import java.nio.file.Files;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public abstract class CanalCdcBase {
    // 通用配置
    protected static final String BOOTSTRAP_SERVERS = "10.94.162.31:9092";
    protected static final String TOPIC = "ucenter_demo";
    protected static final ObjectMapper objectMapper = new ObjectMapper();
    protected final Logger log;

    protected CanalCdcBase() {
        // 使用子类的类名初始化logger
        this.log = LoggerFactory.getLogger(this.getClass());
    }

    // 抽象方法：获取JSON文件路径
    protected abstract String getJsonFilePath();

    // 处理Canal CDC消息的主要逻辑
    protected void processCanalMessages() {
        try {
            // 读取JSON文件
            String jsonContent = new String(Files.readAllBytes(
                new File(getJsonFilePath()).toPath()
            ));
            
            // 解析JSON数组
            List<CanalMessage> messages = Arrays.asList(
                objectMapper.readValue(jsonContent, CanalMessage[].class)
            );
            
            // 获取Kafka生产者配置
            Properties props = KafkaConfig.getProducerProperties(BOOTSTRAP_SERVERS);
            
            // 创建Kafka生产者
            try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
                // 遍历发送消息
                for (CanalMessage message : messages) {
                    sendMessage(producer, message);
                    // 模拟Canal的发送间隔
                    TimeUnit.SECONDS.sleep(2);
                }
                
                // 等待所有消息发送完成
                producer.flush();
                log.info("所有消息发送完成");
            }
            
        } catch (Exception e) {
            log.error("程序执行出错", e);
        }
    }

    // 发送单条消息的方法
    private void sendMessage(KafkaProducer<String, String> producer, CanalMessage message) throws Exception {
        // 打印消息详情
        log.info("准备发送CDC消息:");
        log.info("  - 消息ID: {}", message.getId());
        log.info("  - 操作类型: {}", message.getType());
        
        // 仅处理UPDATE类型的消息
        if ("UPDATE".equals(message.getType())) {
            if (message.getData() != null && !message.getData().isEmpty() && 
                message.getOld() != null && !message.getOld().isEmpty()) {
                
                // 获取变更前后的数据
                Map<String, Object> newData = message.getData().get(0);
                Map<String, String> oldData = message.getOld().get(0);
                
                // 统计实际变更的字段
                int changedFieldCount = 0;
                log.info("  - 字段变更详情:");
                
                // 遍历新数据，与旧数据比较找出变更字段
                newData.entrySet().stream()
                    .sorted(Map.Entry.comparingByKey())
                    .forEach(entry -> {
                        String fieldName = entry.getKey();
                        Object newValue = entry.getValue();
                        String oldValue = oldData.get(fieldName);
                        
                        // 如果字段在old中存在，说明是变更字段
                        if (oldData.containsKey(fieldName)) {
                            log.info("    * {}: {} -> {}", fieldName, oldValue, newValue);
                        }
                    });
                
                // 计算实际变更的字段数
                changedFieldCount = oldData.size();
                log.info("  - 实际变更字段数: {}", changedFieldCount);
            } else {
                log.info("  - 变更字段: 无效的变更数据");
            }
        }

        // 将消息对象转换为JSON字符串
        String messageJson = objectMapper.writeValueAsString(message);
        
        // 创建ProducerRecord，使用表名作为key
        ProducerRecord<String, String> record = 
            new ProducerRecord<>(TOPIC, message.getTable(), messageJson);
        
        // 发送消息
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                log.info("消息发送成功 - Topic: {}, Partition: {}, Offset: {}, 操作类型: {}", 
                    metadata.topic(), metadata.partition(), metadata.offset(), message.getType());
            } else {
                log.error("消息发送失败", exception);
            }
        });
    }
} 